/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package com.eluup.flume.sink.elasticsearch.client;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.elasticsearch.action.ActionRequest;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateRequestBuilder;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentHelper;
import org.elasticsearch.index.VersionType;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.eluup.flume.sink.elasticsearch.ElasticSearchScriptSerializer;
import com.eluup.flume.sink.elasticsearch.ElasticSearchSinkConstants;
import com.eluup.flume.sink.elasticsearch.IndexNameBuilder;
import com.google.common.annotations.VisibleForTesting;

public class ElasticSearchTransportScriptClient implements ElasticSearchClient {

	public static final Logger logger = LoggerFactory.getLogger(ElasticSearchTransportScriptClient.class);

	private InetSocketTransportAddress[] serverAddresses;
	private ElasticSearchScriptSerializer serializer;
	private BulkRequestBuilder bulkRequestBuilder;

	private Client client;
	
	private int retry = 0;
	private int RETRYTIME = 3;

	private int failSize;

	/**
	 * Transport client for external cluster
	 *
	 * @param hostNames
	 * @param clusterName
	 * @param serializer
	 */
	public ElasticSearchTransportScriptClient(String[] hostNames, String clusterName,
	        ElasticSearchScriptSerializer serializer) {
		configureHostnames(hostNames);
		this.serializer = serializer;
		openClient(clusterName);
		verifySerializer();
	}

	public ElasticSearchTransportScriptClient(String[] hostNames, String clusterName) {
		configureHostnames(hostNames);
		openClient(clusterName);
		verifySerializer();
	}

	/**
	 * Local transport client only for testing
	 *
	 * @param serializer
	 */
	public ElasticSearchTransportScriptClient(ElasticSearchScriptSerializer serializer) {
		this.serializer = serializer;
		verifySerializer();
	}

	/**
	 * Used for testing
	 *
	 * @param client
	 *            ElasticSearch Client
	 * @param serializer
	 *            Event Serializer
	 */
	public ElasticSearchTransportScriptClient(Client client, ElasticSearchScriptSerializer serializer) {
		this.client = client;
		this.serializer = serializer;
		verifySerializer();
	}

	private void verifySerializer() {
		if (!(this.serializer instanceof ElasticSearchScriptSerializer)) {
			logger.error("Serialized object type error! Please use [com.eluup.flume.sink.elasticsearch.ElasticSearchScriptSerializer]!");
			System.exit(0);
		}
    }

	@VisibleForTesting
	InetSocketTransportAddress[] getServerAddresses() {
		return serverAddresses;
	}

	@VisibleForTesting
	void setBulkRequestBuilder(BulkRequestBuilder bulkRequestBuilder) {
		this.bulkRequestBuilder = bulkRequestBuilder;
	}

	private void configureHostnames(String[] hostNames) {
		serverAddresses = new InetSocketTransportAddress[hostNames.length];
		for (int i = 0; i < hostNames.length; i++) {
			String[] hostPort = hostNames[i].trim().split(":");
			String host = hostPort[0].trim();
			int port = hostPort.length == 2 ? Integer.parseInt(hostPort[1].trim())
			        : ElasticSearchSinkConstants.DEFAULT_PORT;
			serverAddresses[i] = new InetSocketTransportAddress(new InetSocketAddress(host, port));
		}
	}

	public void close() {
		if (client != null) {
			client.close();
		}
		client = null;
	}

	public void addEvent(Event event, IndexNameBuilder indexNameBuilder, String indexType,
	        long ttlMs) throws Exception {
		if (bulkRequestBuilder == null) {
			bulkRequestBuilder = client.prepareBulk();
		}
		
		XContentBuilder bytesstream = (XContentBuilder) serializer.getContentBuilder(event);
		String pk = serializer.getPrimaryKey();
		String script = serializer.getScript();
		
//		Long version = serializer.getVersion();
		if(pk.isEmpty()) {
			pk = "empty";
		}
		Map<String, ?> params = serializer.getParams();
		UpdateRequestBuilder updateRequestBuilder = client
		        .prepareUpdate(indexNameBuilder.getIndexName(event), indexType, pk)
		        .setUpsert(bytesstream);
		if (script != null) {
			updateRequestBuilder.setScript(new Script(script, ScriptService.ScriptType.INLINE, null, params));
		}
		bulkRequestBuilder.add(updateRequestBuilder);
	}

	public void execute() throws Exception {
		try {
			BulkResponse bulkResponse = bulkRequestBuilder.execute().actionGet();
			
			
			failSize = 0;
			List<ActionRequest<?>> bulkRequestList = bulkRequestBuilder.request().requests();
			bulkRequestBuilder = client.prepareBulk();
			for (BulkItemResponse response : bulkResponse.getItems()) {
            	if (response.isFailed()) {
            		failSize++;
            		int itemId = response.getItemId();
            		UpdateRequest updateRequest  = (UpdateRequest) bulkRequestList.get(itemId);
            		IndexRequest indexRequest = (IndexRequest) updateRequest.upsertRequest();
            		String body = XContentHelper.convertToJson(indexRequest.source(), false);
            		
            		System.out.println("index:" + response.getIndex() + "/" + response.getType() + "/" + response.getId());
            		System.out.println("body:" + body);
            		System.out.println("script:" + updateRequest.script());
            		
            		bulkRequestBuilder.add(client
            		        .prepareUpdate(response.getIndex(), response.getType(), response.getId())
            		        .setUpsert(body)
            		        .setScript(updateRequest.script()));
            		
            		if (retry == RETRYTIME) {
//            			System.out.println("error:====");
    					logger.error("index:" + response.getIndex() + "/" + response.getType() + "/" + response.getId());
    					logger.error("body:" + body);
    					logger.error("script:" + updateRequest.script());
    					System.out.println(bulkResponse.buildFailureMessage());
            		}
	            }
	        }
			System.out.println("failSize:[" + failSize + "/" + bulkResponse.getItems().length + "]");
			if (bulkResponse.hasFailures() && retry < RETRYTIME ) {
				retry++;
				execute();
			}
		} finally {
			retry = 0;
			System.out.println("=============================================");
			//bulkRequestBuilder = client.prepareBulk();
		}
	}

	/**
	 * Open client to elaticsearch cluster
	 *
	 * @param clusterName
	 */
	private void openClient(String clusterName) {
		logger.info("Using ElasticSearch hostnames: {} ", Arrays.toString(serverAddresses));

		Settings settings = Settings.builder().put("cluster.name", clusterName).build();
		if (client != null) {
			client.close();
		}
		PreBuiltTransportClient transportClient = new PreBuiltTransportClient(settings);
		for (InetSocketTransportAddress host : serverAddresses) {
			transportClient.addTransportAddress(host);
		}
		client = transportClient;
	}


	public void configure(Context context) {
		// To change body of implemented methods use File | Settings | File
		// Templates.
	}
	
}
